package cn.edu.gmall.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.io.Serializable;
import java.util.Map;

public class MyHbaseUtil implements Serializable {
    private static final org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();// 配置
    private static Connection connection;

    static {
        config.set("hbase.zookeeper.quorum", "localhost");// zookeeper地址
        config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口
        try {
            connection = ConnectionFactory.createConnection(config);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void makeTable(String name) throws IOException {
        // 创建表管理类
        Admin admin = connection.getAdmin();// hbase表管理
        // 创建表描述类
        TableName tableName = TableName.valueOf(name); // 表名称
        if (!admin.tableExists(tableName)) {
            HTableDescriptor desc = new HTableDescriptor(tableName);
            // 创建列族的描述类
            HColumnDescriptor family = new HColumnDescriptor("fields"); // 列族
            // 将列族添加到表中
            desc.addFamily(family);
            // 创建表
            admin.createTable(desc); // 创建表
            System.out.println("创建表 " + name + " 成功！");
        }
        admin.close();
    }

    public static JSONObject get(String tableName, String id) {
        JSONObject jsonObject = null;
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(id));
            get.addFamily(Bytes.toBytes("fields"));
            Result result = table.get(get);
            Cell[] cells = result.rawCells();

            jsonObject = new JSONObject();
            //遍历具体数据
            for (Cell cell : cells) {
                jsonObject.put(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
            }

            table.close();
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("查询" + tableName + id + "失败");
        }
        return jsonObject;
    }

    public static HBaseSinkFunction getHBaseSinkBySchema() {
        return new HBaseSinkFunction();
    }


    public static class HBaseSinkFunction extends RichSinkFunction<JSONObject> {

        @Override
        public void invoke(JSONObject jsonObject, Context context) throws Exception {
//            System.out.println(jsonObject);
            String sink_table = jsonObject.getString("sink_table");
            String id = jsonObject.getJSONObject("data").getString("id");
            Table table = MyHbaseUtil.connection.getTable(TableName.valueOf(sink_table));
            Put put = new Put(Bytes.toBytes(id));

            for (Map.Entry<String, Object> kv : jsonObject.getJSONObject("data").entrySet()) {
                put.addColumn(Bytes.toBytes("fields"), Bytes.toBytes(kv.getKey()), Bytes.toBytes(kv.getValue().toString()));
            }
            table.put(put);
            table.close();

            //判断如果为更新数据,则删除 Redis 中数据
            if ("update".equalsIgnoreCase(jsonObject.getString("operation"))) {
                DimUtil.deleteCached(sink_table + ":" + id);
            }
        }
    }

    public static void main(String[] args) {
        JSONObject jsonObject = MyHbaseUtil.get("base_trademark", "1");
        System.out.println(jsonObject);
    }
}


